package hook

import (
	"fmt"
	"gitee.com/kudingc/zh_logs/library/kafka"
	log "github.com/sirupsen/logrus"
	"os"
)

var (
	kafkaBrokers []string
	kafkaTopic   string
)

func InitKafkaSubscribe(brokers []string, topic string) {

	kafkaBrokers = brokers
	kafkaTopic = topic
}

type KafKaHook struct {
	kafkaProducer *kafka.Producer
}

func (h *KafKaHook) Levels() []log.Level {
	return log.AllLevels
}

func (h *KafKaHook) Fire(entry *log.Entry) error {
	serialized, err := entry.Logger.Formatter.Format(entry)
	if err != nil {
		fmt.Fprintf(os.Stderr, "Failed to obtain reader, %v\n", err)
	} else {
		h.kafkaProducer.SendMsgSync(string(serialized))
	}
	return nil
}

func NewKafkaHook() *KafKaHook {
	producer := kafka.NewKafkaProducer(kafkaBrokers, kafkaTopic, true)
	return &KafKaHook{kafkaProducer: producer}
}
